<?php

use Swoole\Coroutine\Http\Client;

$conf = '';
include 'db.php';
$server = new Swoole\Server('0.0.0.0', 95);

$server->set([
    'worker_num' => 6,
    'reload_async' => true,
    'max_wait_time' => 61
]);
function createRedisConnection($conf)
{
    // 确保Swoole扩展已启用
    $redis = new Redis();
    $redis->connect($conf['redis']['host'], $conf['redis']['port']);
    $redis->auth($conf['redis']['pass']);
    return $redis;
}

function createDatabaseConnection($conf)
{
    $dsn = "mysql:host={$conf['mysql']['host']}:{$conf['mysql']['port']};dbname={$conf['mysql']['database']};charset=utf8mb4";
    $username = $conf['mysql']['username'];
    $password = $conf['mysql']['password'];

    try {
        $pdo = new PDO($dsn, $username, $password);
        $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
        return $pdo;
    } catch (PDOException $e) {
        logfile("Connection failed: " . $e->getMessage());
        exit;
    }
}

$server->on('WorkerStart', function ($server, $worker_id) use ($conf) {
    $arr = ['load', 'chat', 'online','dau','actordau','payment'];
    $workerName = $arr[$worker_id];
    // 将 $workerName 作为 Worker 属性保存
    $server->users[$worker_id]['pdo'] = createDatabaseConnection($conf);
    $server->users[$worker_id]['redis'] = createRedisConnection($conf);

    if ($worker_id <= $server->setting['worker_num']) {
        recoverProcessingQueue($server->users[$worker_id]['redis'],$workerName);
        $timerId = $server->tick(1000, function ($tid) use ($server, $workerName, $worker_id, $conf) {
            $fun = 'handle'.ucfirst($workerName);
            call_user_func($fun, $server, $conf);
        });
        $server->ticks[$worker_id] = $timerId;

        // 每隔 60 秒心跳
        $timerId = $server->tick(60000, function () use ($server, $worker_id,$conf) {
            $redis = $server->users[$worker_id]['redis'];
            try{
                //$info = $redis->info();
                $response = $redis->ping();
                if (trim($response,'+') !== 'PONG' && $response !== true) {
                    throw new Exception("Unexpected PING response: $response");
                }

                //$connected_clients = $info['connected_clients'] ?? 0;
                //logfile("Redis connected {$worker_id} clients: $connected_clients", 'redis_info');
            }catch (Throwable $e){
                logfile("redis ping failed: " . $e->getMessage());
            }

            $pdo = $server->users[$worker_id]['pdo'];
            try {
                $pdo->query('SELECT 1');
                //logfile('heartbeat SELECT 1', 'error_sql');
            } catch (PDOException $e) {
                logfile("Heartbeat failed, reconnecting: " . $e->getMessage());
                $server->users[$worker_id]['pdo'] = createDatabaseConnection($conf);
            }
        });
        $server->ping_ticks[$worker_id] = $timerId;

    }
});

$server->on('WorkerStop', function ($server, $worker_id) use ($conf) {
    clearTimer($server, $worker_id);
});
$server->on('WorkerExit', function ($server, $worker_id) use ($conf) {
    clearTimer($server, $worker_id);
});

function clearTimer($server, $worker_id)
{
    if (isset($server->users[$worker_id]['redis'])) {
        $server->users[$worker_id]['redis']->close();
    }
    if (isset($server->users[$worker_id]['pdo'])) {
        $server->users[$worker_id]['pdo'] = null;
    }
    if (isset($server->ticks[$worker_id])) {
        $server->clearTimer($server->ticks[$worker_id]);
    }
    if (isset($server->ping_ticks[$worker_id])) {
        $server->clearTimer($server->ping_ticks[$worker_id]);
    }
}
function safeName($name){
    return $name.":processing";
}
function handlePayment($server, $conf){
    safeHandle($server,$conf,'report_payment',1,'baseHandlePayment');//版本不支持0.2等浮点数
    //baseHandlePayment($server,$conf);
}

function baseHandlePayment($server, $conf,$res=''){
    if(!$res){
        $redis = $server->users[$server->worker_id]['redis'];
        $res = $redis->rpop('report_payment');
        if(!$res) return;
    }
    $datas = json_decode($res,true) ?: [];
    if(!$datas){
        logfile('====handlePayment--error:'.$res,'payment');
        return;
    }
    $url = $datas['url'] ?? '';
    $orders = $datas['orders'] ?? [];
    $where = $datas['where'] ?? [];
    if(!$url || !$orders || !$where || !($orders['error'] ?? []) || !($orders['success'] ?? [])){
        logfile('====handlePayment--error2:'.json_encode($datas),'payment');
        return;
    }
    if($url){
        goUrl($url,function ($result) use ($url,$where,$orders,$server, $conf){
            logfile($url.'====result:'.$result,'payment');
            $result = json_decode($result,true);
            $order = $orders['error'] ?? [];
            $status = 'success';
            if ($result['code'] != 1) {
                $status = 'error';
            } else {
                $order = $orders['success'] ?? [];
            }
            update($order,buildWhere($where),'log_order', $server, $conf);
            logfile("====handlePayment--$status:".json_encode($where).':'.json_encode($order),'payment');
        });
    }
}

function handleLoad($server, $conf)
{
    $redis = $server->users[$server->worker_id]['redis'];
    $limit = 30;
    $datas = [];
    while ($limit--) {
        $l = $redis->rpop('report_load');
        if ($l) $datas[] = $l;
    }
    // 处理任务
    if ($datas) {
        $i = 0;
        $load = [];
        foreach ($datas as $data) {
            parse_str(urldecode($data), $data);
            if (!isset($data['actorId']) || $data['actorId'] === 'undefined') $data['actorId'] = 0;
            $load[$i]['tag'] = $data['name'];
            $load[$i]['cid'] = $data['platform'];
            $load[$i]['cpid'] = $data['gameId'] ?? '';
            $load[$i]['ccid'] = $data['channel'];
            $load[$i]['device'] = $data['device'];
            $load[$i]['version'] = $data['version'];
            $load[$i]['actor_id'] = $data['actorId'];
            $load[$i]['actor_name'] = $data['actorName'];
            $load[$i]['account'] = $data['openId'];
            $load[$i]['sid'] = empty($data['sid']) ? 0 : $data['sid'];
            $load[$i]['is_new'] = $data['isNew'];
            $load[$i]['ip'] = $data['ip'];
            $load[$i]['msg'] = $data['msg'] ?? '';
            $load[$i]['load_dt'] = date('Y-m-d H:i:s');
            $i++;
        }
        $res = batchInsert($load, 'log_load', $server, $conf);
    }

}

function handleChat($server, $conf)
{
    $redis = $server->users[$server->worker_id]['redis'];
    $c = $redis->rpop('report_chat');
    if ($c) {
        parse_str(urldecode($c), $d);
        $d['chat_dt'] = date('Y-m-d H:i:s');
        $d['status'] = 1;
        $d['ccid'] = $d['channel'];
        unset($d['channel']);
        $d['cid'] = $d['platform'];
        $d['cpid'] = $d['game_id'];
        $d['channel'] = $d['chat_type'];
        unset($d['counter'],$d['chat_type'], $d['data'], $d['platform'], $d['game_id']);
        insert($d, 'log_chat', $server, $conf);
        // 上报到登录接口
        $url = LOGIN_PAY_URL.'/api/channel/syncChat?platform='.$d['cid'].'&'.http_build_query($d);
        logfile($url,'chat');
        //file_get_contents($url);
        goUrl($url,function ($result) use ($url){
            logfile('====goUrl:'.$result,'chat');
        });
    }
}

function handleOnline($server, $conf)
{
    $redis = $server->users[$server->worker_id]['redis'];
    $o = $redis->rpop('report_online');
    if ($o) {
        parse_str(urldecode($o), $d);
        $d['online_dt'] = date('Y-m-d H:i:s', round($d['online_time']/1000));
        //$d['online_date'] = date('Y-m-d', round($d['online_time']/1000));
        unset($d['counter'],$d['data'],$d['online_time'],$d['ip']);
        insert($d, 'log_online', $server, $conf);
    }
}


function handleDau($server, $conf)
{
    $redis = $server->users[$server->worker_id]['redis'];
    $o = $redis->rpop('report_dau');
    if ($o) {
        parse_str(urldecode($o), $d);
        $fields = ['cid','cpid','ccid','sid','account','actor_id','login_dt','is_new'];
        $data = array_intersect_key($d,array_flip($fields));
        insert($data, 'log_dau', $server, $conf);
    }
}

function handleActordau($server, $conf)
{
    $redis = $server->users[$server->worker_id]['redis'];
    $o = $redis->rpop('report_actor_dau');
    if ($o) {

        parse_str(urldecode($o), $d);
        $fields = ['cid','cpid','ccid','sid','account','actor_id','login_dt','is_new'];
        $data = array_intersect_key($d,array_flip($fields));
        insert($data, 'log_actor_dau', $server, $conf);
    }
}


function buildWhere($where){
    $whereStr = [];
    $values = [];
    foreach ($where as $key => $value){
        $whereStr[] = "`$key` = ?";
        $values[] = $value;
    }
    return [implode(' and ',$whereStr),$values];
}

function batchInsert($rows, $table, $server, $conf)
{
    if ($rows) {
        $pdo = $server->users[$server->worker_id]['pdo'];
        $arr = [];
        foreach ($rows as $row) {
            // 插入数据
            $k = implode(',', array_keys($row));
            $arr[] = implode("','", array_values($row));
        }
        $str = implode("'),('", $arr);
        $sql = "INSERT INTO {$table} ($k) VALUES ('{$str}')";
        $statement = $pdo->prepare($sql);
        try {
            $affected_rows = $statement->execute();
        } catch (PDOException $e) {
            logfile("batchInsert SQL error: " . $e->getMessage());
            $pdo = $server->users[$server->worker_id]['pdo'] = createDatabaseConnection($conf);
            $statement = $pdo->prepare($sql);
            try {
                $affected_rows = $statement->execute();
            } catch (PDOException $e) {
                logfile($sql, 'error_sql');
            }
        }
    }
}

function insert($row, $table, $server, $conf)
{
    if ($row) {
        $pdo = $server->users[$server->worker_id]['pdo'];
        $k = implode(',', array_keys($row));
        $v = implode("','", array_values($row));

        $sql = "INSERT INTO {$table} ($k) VALUES ('{$v}')";
        $statement = $pdo->prepare($sql);
        try {
            $affected_rows = $statement->execute();
        } catch (PDOException $e) {
            logfile("insert SQL error: " . $e->getMessage());
            $pdo = $server->users[$server->worker_id]['pdo'] = createDatabaseConnection($conf);
            $statement = $pdo->prepare($sql);
            try {
                $affected_rows = $statement->execute();
            } catch (PDOException $e) {
                logfile($sql, 'error_sql');
            }
        }
    }
}

function update($data, $where,$table, $server, $conf,$pdo=null)
{
    if ($data) {
        !$pdo && $pdo = $server->users[$server->worker_id]['pdo'];
        $keys = array();
        $values = array();
        foreach ($data as $key => $val) {
            is_array($val) && ($val = json_encode($val, JSON_UNESCAPED_UNICODE));
            $keys[] = "`{$key}`=?";
            $values[] = $val;
        }
        $keys = implode(',', $keys);
        if (is_string($where)) {
            $where = array($where, array());
        }
        $values = array_merge($values, $where[1]);
        $sql = "update {$table} set {$keys} where {$where[0]}";
        $statement = $pdo->prepare($sql);
        try {
            $affected_rows = $statement->execute($values);
        } catch (PDOException $e) {
            logfile("update SQL error: " . $e->getMessage());
            $pdo = $server->users[$server->worker_id]['pdo'] = createDatabaseConnection($conf);
            $statement = $pdo->prepare($sql);
            try {
                $affected_rows = $statement->execute();
            } catch (PDOException $e) {
                logfile($sql.json_encode($values), 'error_sql');
            }
        }
    }
}

function safeHandle($server,$conf,$name,$timeOut=10,$function=null){
    $redis = $server->users[$server->worker_id]['redis'];
    $job = $redis->brpoplpush($name, safeName($name), $timeOut);
    if ($job) {
        try {
            $function && $function($server, $conf,$job);
            $redis->LREM(safeName($name), $job, 0);
        } catch (Throwable $e) {
            logfile("safeHandle {$name} failed: " . $e->getMessage());
            // 可选：延迟后重试，或移到 failed 队列
        }
    }
}

function recoverProcessingQueue($redis, $workerName) {
    $names = ['report_payment'];
    if(!in_array('report_'.strtolower($workerName),$names)) return;
    foreach ($names as $name){
        $processingKey = safeName($name);
        $mainKey       = $name;
        // 获取 processing 队列中所有任务
        $jobs = $redis->lrange($processingKey, 0, -1);
        if (!empty($jobs)) {
            // 将所有任务重新推入主队列（从左边推，保证顺序）
            foreach (array_reverse($jobs) as $job) { // reverse 保持原始顺序
                $redis->lpush($mainKey, $job);
            }
            logfile("Recovered " . count($jobs) . " jobs from $processingKey to $mainKey",'recover');
        }
    }

}


function logfile($message, $pre='error')
{
    $logFile = FILE_PATH.$pre.'_'.date('Ymd').'.log';
    $d = date('Y-m-d H:i:s');
    $message = "[$d] {$message}";
    file_put_contents($logFile, $message . PHP_EOL, FILE_APPEND);
}

function parse($data, $param)
{
    $datas = explode("|", $data);
    $params = explode("|", $param);
    $data = array();
    for($i = 0; $i < count($params); ++$i)
    {
        if(isset($datas[$i]))
        {
            if (in_array($params[$i], ['sid', 'actor_id']) && $datas[$i] === '') $datas[$i] = 0;
            $data[$params[$i]] = $datas[$i];
        }
    }
    return $data;
}

function goUrl($url, $func = null)
{
    go(function () use ($url, $func) {
        $parsedUrl = parse_url($url);
        if (!$parsedUrl || !isset($parsedUrl['host'])) {
            $func && $func('{"code":-1,"msg":"invalid url"}');
            return;
        }
        $host = $parsedUrl['host'] ?? ''; // 主机名
        $port = $parsedUrl['port'] ?? (strtolower($parsedUrl['scheme'] ?? '') === 'https' ? 443 : 80);
        $path = $parsedUrl['path'] ?? '/'; // 路径
        $query = $parsedUrl['query'] ?? ''; // 查询参数
        $fullPath = $path . ($query ? '?' . $query : '');
        $client = new Client($host, $port, $port === 443);
        $client->set([
            'timeout' => 5.0,
        ]);
        $client->setHeaders([
            'Host' => $host,
            'User-Agent' => 'Swoole HTTP Client',
        ]);
        $client->get($fullPath);
        $response = $client->body ?: '{"code":-1,"msg":"empty response"}';

        $statusCode = $client->statusCode;
        // 可选：记录响应日志
        if ($statusCode != 200) {
            logfile(var_export(compact('url', 'response'), true));
        }
        $client->close();
        if ($func) {
            /** @var callable $func */
            $func($response);
        }

    });
}


// 设置空的onReceive回调以避免错误
$server->on('receive', function (Swoole\Server $server, $fd, $from_id, $data) {
    // 这里可以处理接收到的数据
    if($data=='reload'){
        $server->reload(false);
    }
    $server->send($fd, "receive data: $data");
});

$server->start();
